package cn.xuexiyuan.flinkstudy.source;

import cn.xuexiyuan.flinkstudy.entity.Order;
import cn.xuexiyuan.flinkstudy.test.DataFactory;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;


/**
 * @Description:  自定义数据源: 生成随机订单
 *
 * @Author 左龙龙
 * @Date 21-3-23
 * @Version 1.0
 **/
public class SourceDemo04_Costomer {
    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 1.source
        DataStreamSource<Order> orderDS = env.addSource(new MyOrderSource())
                .setParallelism(1);

        // 2.transformation

        // 3.sink
        orderDS.print();

        // 4.excute
        env.execute("SourceDemo04_Costomer");
    }
}

class MyOrderSource extends RichParallelSourceFunction<Order>{

    @Override
    public void run(SourceContext<Order> sourceContext) throws Exception {
        while (true){
            sourceContext.collect(DataFactory.createRandomOrder(100, 3));

            Thread.sleep(1000);
        }
    }

    // 执行 cancel 的时候执行
    @Override
    public void cancel() {

    }
}
